package com.aliyun.iot.demo.iothub;

import java.io.UnsupportedEncodingException;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.aliyun.iot.demo.iothub.DataProvider.Sample;
import com.aliyun.iot.util.LogUtil;

public class ConnectMqtt implements Runnable {

	private static MqttClient sampleClient;
	private static Thread t;

	public static String deviceName = "demo0";
	public static String productKey = "a1l7anRRgND";
	public static String secret = "XKOtD4bcso3GuaorNJOrzyCZ8avaicPD";

	private static ObjectActionDefine head;

	// 用于测试的topic
	private static String pubTopic0 = "/" + productKey + "/" + deviceName + "/alarm0";
	private static String pubTopic1 = "/" + productKey + "/" + deviceName + "/alarm1";
	private static String pubTopic2 = "/" + productKey + "/" + deviceName + "/alarm2";

	public ConnectMqtt(String url, String clientId, String mqttUsername, String mqttPassword, String deviceName,
			ObjectActionDefine head) throws Exception {
		this.head = head;
		MemoryPersistence persistence = new MemoryPersistence();
		SSLSocketFactory socketFactory = createSSLSocket();
		sampleClient = new MqttClient(url, clientId, persistence);
		MqttConnectOptions connOpts = new MqttConnectOptions();
		connOpts.setMqttVersion(4); // MQTT 3.1.1
		connOpts.setSocketFactory(socketFactory);

		// 设置是否自动重连
		connOpts.setAutomaticReconnect(true);

		// 如果是true，那么清理所有离线消息，即QoS1或者2的所有未接收内容
		connOpts.setCleanSession(false);

		connOpts.setUserName(mqttUsername);
		connOpts.setPassword(mqttPassword.toCharArray());
		connOpts.setKeepAliveInterval(65);

		LogUtil.print(clientId + "进行连接, 目的地: " + url);
		sampleClient.connect(connOpts);

		sampleClient.setCallback(new MqttCallback() {
			@Override
			public void connectionLost(Throwable cause) {
				LogUtil.print("连接失败,原因:" + cause);
				cause.printStackTrace();
			}

			@Override
			public void messageArrived(String topic, MqttMessage message) throws Exception {
				LogUtil.print(
						"接收到消息,来至Topic [" + topic + "] , 内容是:[" + new String(message.getPayload(), "UTF-8") + "],  ");
			}

			@Override
			public void deliveryComplete(IMqttDeliveryToken token) {
				// 如果是QoS0的消息，token.resp是没有回复的
				LogUtil.print("消息发送成功! "
						+ ((token == null || token.getResponse() == null) ? "null" : token.getResponse().getKey()));
			}
		});
		LogUtil.print("连接成功:---");
		if (t == null) {
			t = new Thread(this, "send message");
			t.start();
		}
	}

	private static SSLSocketFactory createSSLSocket() throws Exception {
		SSLContext context = SSLContext.getInstance("TLSV1.2");
		context.init(null, new TrustManager[] { new ALiyunIotX509TrustManager() }, null);
		SSLSocketFactory socketFactory = context.getSocketFactory();
		return socketFactory;
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		System.out.println("数据判断");

		boolean flag1_last = true;
		float flag1_lastf = 0;
		int flag1_len = 0;
		int flag1_skip = 0;
		while (true) {
			DataProvider.Sample SampleData = head.GetProvider() != null ? head.GetProvider().fetch()
					: DataProvider.Sample.Zero;

			float flag2 = SampleData.Acc[0] * SampleData.Acc[0] + SampleData.Acc[1] * SampleData.Acc[1]
					+ SampleData.Acc[2] * SampleData.Acc[2];

			float flag0 = SampleData.Angle[0];
			// System.out.println("X: "+ SampleData.Angle[0] + " Y: " + SampleData.Angle[1]
			// + " Z: " + SampleData.Angle[2]);

			if (flag1_skip <= 4) {
				flag1_skip += 1;
			} else {
				flag1_skip = 0;
				
				if (0 == flag1_len) {
					flag1_last = flag1_lastf < flag2;
					flag1_len = 1;
				} else {
					boolean cur = flag1_lastf < flag2 && Math.abs(flag1_lastf - flag2) > 0.02;

					if (cur != flag1_last) {
						flag1_last = cur;
						flag1_len += 1;
					} else {
						flag1_len = 0;
					}
				}
				
				flag1_lastf = flag2;
			}

			try {
				Thread.sleep(50);
			} catch (Exception e) {
				// TODO: handle exception
			}

			if (flag1_skip == 0) {
				System.out.println("flag1_len: " + flag1_len + ", flag2: " + flag2);
			}
			if (flag1_len > 5) {
				flag1_len = 0;
				System.out.println("发消息啦：癫痫");
				System.out.println("flag1_len:" + flag1_len);
				String content = "{'content':'msg : epilepsy" + "'}";
				MqttMessage message;
				try {
					message = new MqttMessage(content.getBytes("utf-8"));
					message.setQos(0);
					sampleClient.publish(pubTopic1, message);
					Thread.sleep(5000);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			} else if (Math.abs(flag0) < 15) {
				System.out.println("发消息啦：心脏病");
				System.out.println("flag0:" + flag0);
				String content = "{'content':'msg : fallen down" + "'}";
				MqttMessage message;
				try {
					message = new MqttMessage(content.getBytes("utf-8"));
					message.setQos(0);
					sampleClient.publish(pubTopic0, message);
					Thread.sleep(5000);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			} else if (2.2 <= flag2) {
				System.out.println("发消息啦:冲击");
				System.out.println("flag2:" + flag2);
				String content = "{'content':'msg : calapse" + "'}";
				MqttMessage message;
				try {
					message = new MqttMessage(content.getBytes("utf-8"));
					message.setQos(0);
					sampleClient.publish(pubTopic2, message);
					Thread.sleep(5000);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}
}
